﻿using System;
using System.Threading.Tasks;
using System.Collections.Generic;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.Exceptions;

namespace YEA.Infrastructure.Gateway
{
    public class AsyncConsumerProcessor : IAsyncConsumerProcessor 
    {
        public string Queue { get; set; }
        protected MessageConsumer _handler;
        protected IConnection _connection;
        protected List<IModel> _channels;
        protected List<Task> _tasks;
        protected int _maxthreads;

        public AsyncConsumerProcessor(IConnection connection, int maxthreads, MessageConsumer handler, string queue)
        {
            if (connection == null)
                throw new ApplicationException("The connection parameter is missing.  A connection is required to publish messages");
            if (handler == null)
                throw new ApplicationException("The handler is missing and is required to process messages received from the message bus");
            if (string.IsNullOrWhiteSpace(queue))
                throw new ApplicationException("A valid queue name has to be specified");
            if (maxthreads <= 0)
                throw new ApplicationException("The asnynchronous listener has to have more than one thread to listen to");
            _maxthreads = maxthreads == 0 ? 1 : maxthreads;
            Queue = queue;
            _connection = connection;
            _handler = handler;
            _channels = new List<IModel>();
            _tasks = new List<Task>();
            try
            {
                for (int i = 0; i < _maxthreads; i++)
                {
                    var channel = _connection.CreateModel();
                    _channels.Add(channel);
                    var task = Task.Factory.StartNew(() => MessageProcessor.MessageReceiver(channel, Queue, _handler));
                    _tasks.Add(task);
                }

                Task.WaitAll(_tasks.ToArray());
            }
            catch (AggregateException ae)
            {
                //Todo: do something with exceptions that can be solved.
                throw ae.Flatten();
            }
        }

        public void Stop()
        {
            Parallel.ForEach(_channels, (channel) =>
            {
                channel.Close(2, "Basic Stop Request"); 
                channel.Dispose();
            });
        }
    }
    public static class MessageProcessor
    {
        public static void MessageReceiver(IModel channel, string queue, MessageConsumer handler)
        {
            var consumer = new QueueingBasicConsumer(channel);
            var consumerTag = string.Empty;
            foreach (var onequeue in queue.Split(','))
            {
                consumerTag = channel.BasicConsume(onequeue.Trim(), false, consumer);
            }
            while (true)
            {
                var deliveryArgs = consumer.Queue.Dequeue() as BasicDeliverEventArgs;
                var message = new Message()
                {
                    Properties = deliveryArgs.BasicProperties,
                    Body = deliveryArgs.Body,
                    RoutingKey = deliveryArgs.RoutingKey,
                };
                DeliveryHeader header = new DeliveryHeader()
                {
                    Redelivered = deliveryArgs.Redelivered,
                    DeliveryTag = deliveryArgs.DeliveryTag
                };
                handler(message, channel, header);
            }

        }


    }
}
